package com.lagou.demo6;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

public class ToKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .withBuiltInCatalogName("default-catalog")
                .withBuiltInDatabaseName("default-database")
                .build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        DataStreamSource<String> data = env.addSource(new SourceFunction<String>() {
            int count = 0;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (true) {
                    ctx.collect("animal" + count);
                    count++;
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        Table nameTable = tEnv.fromDataStream(data, $("name"));

        tEnv.connect(new Kafka()
                .version("universal")
                .topic("animal")
                .startFromEarliest()
                .property("bootstrap.servers","zb26105:9092")

        )
                .withFormat(new Csv())
                .withSchema(
                        new Schema().field("name", DataTypes.STRING())
                )
                .createTemporaryTable("animalTable");


        nameTable.executeInsert("animalTable");

//        env.execute();   //不需要再使用execute()方法。运行execute()方法时，没有待执行的sql语句，因此会抛出此错误。
                           //错误为：Exception in thread "main" java.lang.IllegalStateException: No operators defined in streaming topology. Cannot generate StreamGraph.


    }

}
